/*
 * Copyright Hundsun Technologies Inc. All Rights Reserved.
 */

package miner

import (
	"context"
	"math/rand"
	"os"
	"sync"
	"sync/atomic"
	"time"

	"hundsun.com/hsl/hschain/common/log/logging"
	"hundsun.com/hsl/hschain/protos/consensus/raft"
)

// CMState 共识节点状态
type CMState int

const (
	// Follower follower节点
	Follower CMState = iota
	// Candidate candidate节点
	Candidate
	// Leader leader节点
	Leader
	// Dead dead节点
	Dead
)

// String 节点状态转string
func (s CMState) String() string {
	switch s {
	case Follower:
		return "Follower"
	case Candidate:
		return "Candidate"
	case Leader:
		return "Leader"
	case Dead:
		return "Dead"
	default:
		panic("unreachable")
	}
}

// ConsensusModule (CM) 实现的是Raft一致性集群中的一个节点
type ConsensusModule struct {
	// mu 用于保护对CM的并发访问.
	mu sync.Mutex

	// id 当前CM对应的服务器ID.
	id int64

	// peerIds 罗列了集群中所有同伴的ID.
	peerIds []int64
	clients []*RaftConsensusClient

	// 所有服务器都需要持久化存储的 Raft state
	currentTerm int64
	votedFor    int64

	// 所有服务器中经常修改的 Raft state
	state              CMState
	electionResetEvent time.Time

	// triggerCommitChan是一个内部通知channel，触发心跳向追随者发送提交请求。
	triggerCommitChan chan struct{}
	// 心跳服务完成收集追随者的提交请求后，
	commitReadyChan chan struct{}

	// 序列号中心的最新块序号，用来表示当前轮次，避免请求提交时领导者收到上一轮次的回应造成混淆
	lastBlockSequence uint64
	// 同意提交请求的peer数量计数
	count     int
	committed bool

	log logging.Log
	ctx context.Context
}

// NewConsensusModule 方法使用给定的服务器ID、同伴ID列表peerIds以及服务器server来创建一个新的CM实例。
// ready channel用于告知CM所有的同伴都已经连接成功，可以安全启动状态机。
func NewConsensusModule(ctx context.Context, id int64, peerIds []int64, log logging.Log) *ConsensusModule {
	cm := new(ConsensusModule)
	cm.id = id
	cm.peerIds = peerIds
	cm.state = Follower
	cm.votedFor = -1
	cm.log = log
	cm.ctx = ctx
	cm.triggerCommitChan = make(chan struct{})
	cm.commitReadyChan = make(chan struct{})
	return cm
}

// Start 启动ConsensusModule
func (cm *ConsensusModule) Start() {
	go func() {
		cm.mu.Lock()
		cm.electionResetEvent = time.Now()
		cm.mu.Unlock()
		cm.runElectionTimer()
	}()
}

// Report 方法会呈报当前CM的状态属性
func (cm *ConsensusModule) Report() (id int64, term int64, isLeader bool) {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	return cm.id, cm.currentTerm, cm.state == Leader
}

// IsLeader 是否为leader
func (cm *ConsensusModule) IsLeader() bool {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	return cm.state == Leader
}

// Stop 方法可以暂停当前CM，清除其状态。该方法很快结束，
// 但是所有的goroutine退出可能需要一点时间（取决于选举等待时间）
// 提供该接口，当miner检测系统不满足出块条件时主动退出
func (cm *ConsensusModule) Stop() {
	cm.mu.Lock()
	defer cm.mu.Unlock()
	cm.state = Dead
	cm.log.Info("becomes Dead")
}

// Restart 重启cm模块
func (cm *ConsensusModule) Restart() {
	go func() {
		cm.mu.Lock()
		cm.electionResetEvent = time.Now()
		cm.mu.Unlock()
		cm.runElectionTimer()
	}()
}

// electionTimeout方法生成一个伪随机的选举等待时长
func (cm *ConsensusModule) electionTimeout() time.Duration {
	// 如果设置了 RAFT_FORCE_MORE_REELECTION, 会有意地经常返回硬编码的数字来进行压力测试。
	// 这会造成不同服务器之间的冲突，迫使集群出现更多的重新选举
	if len(os.Getenv("RAFT_FORCE_MORE_REELECTION")) > 0 && rand.Intn(3) == 0 {
		return time.Duration(150) * time.Millisecond
	}
	return time.Duration(150+rand.Intn(150)) * time.Millisecond
}

/*
	runElectionTimer实现的是选举定时器。如果我们想在新一轮选举中作为候选人，就要启动这个定时器。
	该方法是阻塞的，需要在独立的goroutine中运行；它应该用作单次（一次性）选举定时器，
	因为一旦任期变化或者CM状态不是追随者/候选人，该方法就会退出。
*/
func (cm *ConsensusModule) runElectionTimer() {
	timeoutDuration := cm.electionTimeout()
	cm.mu.Lock()
	termStarted := cm.currentTerm
	cm.mu.Unlock()
	cm.log.Infof("election timer started (%v), term=%d", timeoutDuration, termStarted)

	/*
	  循环会在以下条件结束：
	  1 - 发现不再需要选举定时器
	  2 - 选举定时器超时，CM变为候选人
	  对于追随者而言，定时器通常会在CM的整个生命周期中一直在后台运行。
	*/
	ticker := time.NewTicker(10 * time.Millisecond)
	defer ticker.Stop()
	for {
		<-ticker.C

		cm.mu.Lock()
		if cm.state != Candidate && cm.state != Follower {
			cm.log.Infof("in election timer state=%s, bailing out", cm.state)
			cm.mu.Unlock()
			return
		}

		if termStarted != cm.currentTerm {
			cm.log.Infof("in election timer term changed from %d to %d, bailing out", termStarted, cm.currentTerm)
			cm.mu.Unlock()
			return
		}

		// 如果在等待时间内没有收到领导者的心跳，也没有为其它候选人投票，就发起新一轮选举。
		if elapsed := time.Since(cm.electionResetEvent); elapsed >= timeoutDuration {
			cm.startElection()
			cm.mu.Unlock()
			return
		}
		cm.mu.Unlock()
	}
}

// startElection方法会将该CM作为候选人发起新一轮选举，要求cm.mu被锁定
func (cm *ConsensusModule) startElection() {
	cm.state = Candidate
	cm.currentTerm++
	savedCurrentTerm := cm.currentTerm
	cm.electionResetEvent = time.Now()
	cm.votedFor = cm.id
	cm.log.Infof("becomes Candidate (currentTerm=%d)", savedCurrentTerm)

	var votesReceived int32 = 1

	for _, peerID := range cm.peerIds {
		go func(peerId int64) {
			client := cm.clients[peerId]
			args := raft.RequestVoteArgs{
				Term:        savedCurrentTerm,
				CandidateId: cm.id,
			}
			cm.log.Infof("sending RequestVote to %d: %v", peerId, args)
			if reply, err := client.pbClient.RequestVote(cm.ctx, &args); err == nil {
				cm.mu.Lock()
				defer cm.mu.Unlock()
				cm.log.Infof("received RequestVoteReply %v", reply)

				// 状态不是候选人，退出选举（可能退化为追随者，也可能已经胜选成为领导者）
				if cm.state != Candidate {
					cm.log.Infof("while waiting for reply, state = %v", cm.state)
					return
				}

				// 存在更高任期（新领导者），转换为追随者
				if reply.Term > savedCurrentTerm {
					cm.log.Infof("term out of date in RequestVoteReply")
					cm.becomeFollower(reply.Term)
					return
				} else if reply.Term == savedCurrentTerm {
					if reply.VoteGranted {
						votes := int(atomic.AddInt32(&votesReceived, 1))
						if votes*2 > len(cm.peerIds)+1 {
							// 获得票数超过一半，选举获胜，成为最新的领导者
							cm.log.Infof("wins election with %d votes", votes)
							cm.startLeader()
							return
						}
					}
				}
			}
		}(peerID)
	}

	// 另行启动一个选举定时器，以防本次选举不成功
	go cm.runElectionTimer()
}

// becomeFollower方法将cm变为追随者并重置其状态。要求cm.mu锁定
func (cm *ConsensusModule) becomeFollower(term int64) {
	cm.log.Infof("becomes Follower with term=%d", term)
	cm.state = Follower
	cm.currentTerm = term
	cm.votedFor = -1
	cm.electionResetEvent = time.Now()

	go cm.runElectionTimer()
}

// startLeader方法将CM转换为领导者，并启动心跳程序。要求cm.mu锁定
func (cm *ConsensusModule) startLeader() {
	cm.state = Leader
	/*
		该goroutine在后台运行并向同伴服务器发送AE请求：
		- triggerAEChan通道发送任何内容时
		- 如果triggerAEChan通道没有内容时，每50ms执行一次
	*/
	go func(heartbeatTimeout time.Duration) {
		// 立即发送心跳
		cm.leaderSendAEs(false)

		t := time.NewTimer(heartbeatTimeout)
		defer t.Stop()
		for {
			doSend := false
			doCommitRequest := false
			select {
			// 定时器到期
			case <-t.C:
				doSend = true
				// 重置定时器
				t.Stop()
				t.Reset(heartbeatTimeout)
			case _, ok := <-cm.triggerCommitChan:
				if ok {
					doSend = true
					// 需要发送commit请求给跟随者
					doCommitRequest = true
				} else {
					return
				}

				// 重置定时器
				if !t.Stop() {
					<-t.C
				}
				t.Reset(heartbeatTimeout)
			}

			if doSend {
				cm.mu.Lock()
				if cm.state != Leader {
					cm.mu.Unlock()
					return
				}
				cm.mu.Unlock()
				cm.leaderSendAEs(doCommitRequest)
			}
		}
	}(50 * time.Millisecond)
}

// leaderSendAEs定时发送心跳信息, 当commitRequest为true时，附带提交请求信息
func (cm *ConsensusModule) leaderSendAEs(commitRequest bool) {
	cm.mu.Lock()
	savedCurrentTerm := cm.currentTerm
	cm.mu.Unlock()

	for _, peerID := range cm.peerIds {
		go func(peerId int64) {
			cm.mu.Lock()
			client := cm.clients[peerId]
			args := raft.CommitRequestArgs{
				Term:              savedCurrentTerm,
				LeaderId:          cm.id,
				BeingCommit:       commitRequest,
				LastBlockSequence: cm.lastBlockSequence,
			}
			cm.mu.Unlock()
			if commitRequest {
				cm.log.Infof("sending CommitRequest to %d: %v", peerId, args)
			}
			if reply, err := client.pbClient.CommitRequest(cm.ctx, &args); err == nil {
				cm.mu.Lock()
				defer cm.mu.Unlock()
				// 存在更高任期，转换为追随者
				if reply.Term > savedCurrentTerm {
					cm.log.Infof("term out of date in leaderSendAEsReply")
					cm.becomeFollower(reply.Term)
					return
				}

				if cm.state == Leader && savedCurrentTerm == reply.Term {
					if reply.Success {
						// 检查当前块是否可以提交，用序列号保证只有本轮的回复有效
						if reply.LastBlockSequence == cm.lastBlockSequence {
							cm.count++
						}
						// 有可提交指令，通知集群
						if cm.count*2 >= len(cm.peerIds)+1 {
							// 超过半数同意commit，如果出现网络分区，最多也只有一个leader能commit
							// 通过commit channel向miner发送信号，收到该信号，miner可以进行提交，即将块提交给序列号中心
							// 保证只发送一次信号
							if !cm.committed {
								cm.committed = true
								cm.commitReadyChan <- struct{}{}
							}
						}
					}
				}
			}
		}(peerID)
	}
}

// Commit 请求提交，领导者向跟随者发送提交请求，通过后才能发送块到序列中心
// TODO:超时机制，如果一段时间未收到commitReady信号，则放弃
func (cm *ConsensusModule) Commit(blockSequence uint64) bool {
	cm.lastBlockSequence = blockSequence
	cm.mu.Lock()
	cm.log.Infof("%v ...Committing", cm.state)
	if cm.state != Leader {
		cm.mu.Unlock()
		return false
	}
	// 请求提交，计数器记为0
	cm.count = 0
	cm.committed = false
	cm.mu.Unlock()
	// 通知心跳向跟随者发送commit请求
	cm.triggerCommitChan <- struct{}{}
	// 等待commit确认成功
	<-cm.commitReadyChan
	return true
}

// AddNode 增加节点
// 为了避免出现网络分区的问题只能一次增加或删除一个节点
// Raft论文中配置变更
func (cm *ConsensusModule) AddNode() {}

// DeleteNode 删除节点
func (cm *ConsensusModule) DeleteNode() {}
